﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using PServiceBus.Services.Gateway.Runtime;
using PServiceBus.Core.Interface;
using PServiceBus.Core.Runtime.Extensions;
using Phoenix.ReflectionExtensions;
using System.Threading;
using System.Collections.Concurrent;
using PServiceBus.MessageProxy.Interface;
using System.Reflection;
using System.Linq.Expressions;
using System.Runtime.InteropServices;
using System.Diagnostics;
using System.Threading.Tasks;

namespace PServiceBus.MessageProxy {
    public class ESBHelper {

        private static ConcurrentDictionary<string, dynamic> _listeners =
            new ConcurrentDictionary<string, dynamic>();

        private static ConcurrentDictionary<string, MethodInfo> _methods =
            new ConcurrentDictionary<string, MethodInfo>();

        private static ConcurrentDictionary<string, Guid> _guidKeys =
            new ConcurrentDictionary<string, Guid>();

        private static ConcurrentDictionary<Guid, object> _responses =
            new ConcurrentDictionary<Guid, object>();

        private static ConcurrentDictionary<Guid, Exception> _exceptions =
            new ConcurrentDictionary<Guid, Exception>();

        private static ConcurrentDictionary<Guid, Subscriber> _subscribers =
            new ConcurrentDictionary<Guid, Subscriber>();

        private static ConcurrentDictionary<Guid, ManualResetEventSlim> _resets =
            new ConcurrentDictionary<Guid, ManualResetEventSlim>();

        private static ConcurrentDictionary<Guid, dynamic> _handlers =
            new ConcurrentDictionary<Guid, dynamic>();

        private static ESBHelper _instance;

        private static Subscriber _subscriber;

        private static Type _type = typeof(ESBHelper);

        public static ESBHelper Instance {
            get {
                return _instance ?? (_instance = new ESBHelper());
            }
        }

        private static string GetUniqueSubscriberName() {
            var process = Process.GetCurrentProcess();
            var subname = String.Format("{0}{1}{2}",
                process.ProcessName, process.MachineName,
                AppDomain.CurrentDomain.BaseDirectory.Replace("/", string.Empty)
                    .Replace("\\", string.Empty).Replace(" ", string.Empty));
            return subname;
        }

        private static void CreateSubscriber() {
            if (_subscriber != null) return;
            _subscriber = Subscriber.SelectOrAdd(GetUniqueSubscriberName());
        }

        internal static void Initialize() {
            CreateSubscriber();
        }

        static ESBHelper() {
            Console.CancelKeyPress += Shutdown;
            AppDomain.CurrentDomain.DomainUnload += Shutdown;
            AppDomain.CurrentDomain.ProcessExit += Shutdown;
            _consoleHandler += new EventHandler(Handler);
            SetConsoleCtrlHandler(_consoleHandler, true);
        }

        [DllImport("Kernel32")]
        private static extern bool SetConsoleCtrlHandler(EventHandler handler, bool add);

        private delegate bool EventHandler(CtrlType sig);
        private static EventHandler _consoleHandler;

        private enum CtrlType {
            CTRL_C_EVENT = 0,
            CTRL_BREAK_EVENT = 1,
            CTRL_CLOSE_EVENT = 2,
            CTRL_LOGOFF_EVENT = 5,
            CTRL_SHUTDOWN_EVENT = 6
        }

        private static bool Handler(CtrlType sig) {
            switch (sig) {
                case CtrlType.CTRL_C_EVENT:
                case CtrlType.CTRL_LOGOFF_EVENT:
                case CtrlType.CTRL_SHUTDOWN_EVENT:
                case CtrlType.CTRL_CLOSE_EVENT:
                    ESBHelper.Instance.Shutdown();
                    return false;
                default:
                    return false;
            }
        }

        private static void Shutdown(object sender, EventArgs e) {
            ESBHelper.Instance.Shutdown();
        }

        public static MethodInfo Select(string name) {
            return _methods.GetOrAdd(name, key => _type.GetMethod(key));
        }

        public void Setup(string host, int port) {
            ESB.Connect(host, port);
        }

        public object Translate(object value) {
            return value.Translate();
        }

        public void Shutdown() {
            var handlers = _handlers.Select(x => x.Value)
                .Union(_listeners.Select(x => x.Value));

            foreach (var handler in handlers) {
                handler.Stop();
            }

            if (_subscriber != null && _subscriber.IsValid) _subscriber.Delete();
            foreach (var kv in _subscribers) {
                var subscriber = kv.Value;
                if (subscriber != null && subscriber.IsValid) 
                    subscriber.Delete();
            }

            _subscriber = null;
            _handlers.Clear();
            _listeners.Clear();
            _subscribers.Clear();
        }

        public void EndListener(string id, string topicName) {
            var key = id + topicName;
            if (!_listeners.Any(x => x.Key == key)) return;
            dynamic handler = null;
            if (!_listeners.TryRemove(key, out handler)) return;
            if (_subscriber.IsTransportRegistered(key))
                _subscriber.DeleteTransport(key);
            handler.Stop();
        }

        public void StartListener<TMessage>(int interval, string id, string topicName, ITransport transport, Action<TMessage> action, Func<Exception, bool> func) where TMessage : class {

            if (_subscriber == null)
                throw new InvalidOperationException("ProxyService.Start must be called before invoking listener");

            var key = id + topicName;

            if (_listeners.Any(x => x.Key == key)) 
                throw new InvalidOperationException(String.Format("A listener with the id = '{0}' already exist for {1}!",
                    id, topicName));

            var transportName = key;

            Topic.RegisterIfNotExist(topicName);

            if (interval <= 0) interval = 100;

            if (!_subscriber.IsTransportRegistered(transportName)) {

                if (!_subscriber.IsSubscribeTo(topicName))
                    _subscriber
                        .SubscribeTo(Topic.Select(topicName)).Save();

                _subscriber
                    .AddTransport(transportName, transport, topicName)
                    .Save();
            }


            if (!_listeners.ContainsKey(key)) {
                _subscriber = Subscriber.Select(_subscriber.SubscriberName);
                _listeners[key] = 
                    _subscriber.OnMessageReceived<TMessage>(transportName, action, 
                    error => {
                        error.Continue = func(error.Error);
                    }, interval: TimeSpan.FromMilliseconds(interval));
            }
        }

        public ITransport GetTransport(MethodInfo method) {
            var transportAttr = method.GetAttribute<ESBTransportAttribute>();
            var transportParamAttrs = method.GetAttributes<ESBTransportParamAttribute>();
            return GetTransportByAttributes(transportAttr, transportParamAttrs);
        }

        private ITransport GetTransportByAttributes(ESBTransportAttribute transportAttr, IEnumerable<ESBTransportParamAttribute> transportParams) {
            if (transportAttr == null) return TcpHelper.Transport;
            transportParams = transportParams ?? new List<ESBTransportParamAttribute>();
            var transportType = transportAttr.TransportType;
            var transport = Activator.CreateInstance(transportType) as ITransport;
            foreach (var param in transportParams) {
                var prop = transportType.GetPropertyEx(param.Name);
                if (prop == null) continue;
                prop.SetValueEx(transport, param.Value.ChangeType(prop.PropertyType));
            }
            return transport;
        }

        private static string GetMethodKey(MethodBase method) {
            var type = method.DeclaringType;
            var parameters = method.GetParameters();
            return String.Format("{0}{1}{2}", type.FullName, method.Name,
                String.Join(string.Empty, parameters.Select(x => x.ParameterType.FullName)));
        }

        public TResponse SendWithRequest<TRequest, TResponse>(string topic, ITransport transport, int timeout, TRequest request) where TResponse : class where TRequest : ICorrelation {
            if (request == null) throw new ArgumentNullException("request");
            var stack = new StackTrace(false);
            var key = GetMethodKey(stack.GetFrame(1).GetMethod());
            var correlationID = _guidKeys.GetOrAdd(key, _ => Guid.NewGuid());
            request.ESBCorrelationID = correlationID;
            return SendRequest<TResponse>(topic, transport, timeout, tp => tp.Message(request), request.ESBCorrelationID);
        }

        public Task<TResponse> SendWithRequestTask<TRequest, TResponse>(string topic, ITransport transport, int timeout, TRequest request)
            where TResponse : class
            where TRequest : ICorrelation {
            if (request == null) throw new ArgumentNullException("request");
            var stack = new StackTrace(false);
            var key = GetMethodKey(stack.GetFrame(1).GetMethod());
            var correlationID = _guidKeys.GetOrAdd(key, _ => Guid.NewGuid());
            request.ESBCorrelationID = correlationID;
            return SendRequestTask<TResponse>(topic, transport, timeout, tp => tp.Message(request), request.ESBCorrelationID);
        }

        public TResponse SendWithDict<TResponse>(string topic, ITransport transport, int timeout, Dictionary<string, object> dict) where TResponse : class {
            if (dict == null) throw new ArgumentNullException("dict");
            var stack = new StackTrace(false);
            var key = GetMethodKey(stack.GetFrame(1).GetMethod());
            var correlationID = _guidKeys.GetOrAdd(key, _ => Guid.NewGuid());
            dict[ICorrelation.CORRELATION_ID_KEY] = correlationID;
            return SendRequest<TResponse>(topic, transport, timeout, tp =>
            {
                foreach (var param in dict) 
                    tp.SetParameter(param.Key, param.Value);
            }, (Guid)dict[ICorrelation.CORRELATION_ID_KEY]);
        }

        public Task<TResponse> SendWithDictTask<TResponse>(string topic, ITransport transport, int timeout, Dictionary<string, object> dict) where TResponse : class {
            if (dict == null) throw new ArgumentNullException("dict");
            var stack = new StackTrace(false);
            var key = GetMethodKey(stack.GetFrame(1).GetMethod());
            var correlationID = _guidKeys.GetOrAdd(key, _ => Guid.NewGuid());
            dict[ICorrelation.CORRELATION_ID_KEY] = correlationID;
            return SendRequestTask<TResponse>(topic, transport, timeout, tp =>
            {
                foreach (var param in dict)
                    tp.SetParameter(param.Key, param.Value);
            }, (Guid)dict[ICorrelation.CORRELATION_ID_KEY]);
        }

        public void SendOneWayDict(string topic, Dictionary<string, object> dict) {
            Send(topic, tp =>
            {
                foreach (var param in dict)
                    tp.SetParameter(param.Key, param.Value);
            });
        }

        public void SendOneWayRequest<TRequest>(string topic, TRequest request) {
            Send(topic, tp => tp.Message(request));
        }

        private void Send(string topic, Action<Topic> publishAction) {
            var tp = Topic.Select(topic);
            publishAction(tp);
            tp.Publish();
        }

        private static void SetupRequestResponse<TResponse>(string topic, ITransport transport, Guid correlationID) where TResponse : class {
            
            Topic.RegisterIfNotExist(typeof(TResponse).Name);

            Topic.RegisterIfNotExist(topic);

            var transportName = correlationID.ToString();

            var subscriber = _subscribers.GetOrAdd(correlationID, _ =>
            {
                var subName = _guidKeys.Where(x => x.Value == correlationID).Select(x => x.Key).FirstOrDefault();
                subName += GetUniqueSubscriberName();

                var sub = Subscriber.SelectOrAdd(subName);
                sub.SubscribeTo(Topic.Select<TResponse>().Filter(
                  FilterExpression.Equal(ICorrelation.CORRELATION_ID_KEY, correlationID)))
                  .AddTransportFor<TResponse>(transportName, transport);
                sub.Save();

                return sub;
            });


            if (!_handlers.ContainsKey(correlationID)) {
                _handlers[correlationID] =
                    subscriber.OnMessageReceived<TResponse>(transportName,
                    msg =>
                    {
                        _responses[correlationID] = msg;
                        _resets[correlationID].Set();
                    }, error =>
                    {
                        error.Continue = true;
                        _exceptions[correlationID] = error.Error;
                        _resets[correlationID].Set();
                    });
            }
        }

        private static Task<TResponse> SendRequestTask<TResponse>(string topic, ITransport transport, int timeout, Action<Topic> publishAction, Guid correlationID) where TResponse : class{

            if (_subscriber == null)
                throw new InvalidOperationException("ProxyService.Start must be called before request/response can be invoked");

            SetupRequestResponse<TResponse>(topic, transport, correlationID);

            return new Task<TResponse>(() =>
            {

                var response = default(TResponse);
                var exception = _exceptions[correlationID] = default(Exception);

                _responses[correlationID] = response;
                _resets[correlationID] = new ManualResetEventSlim(false);

                if (timeout <= 0) timeout = -1;

                var tp = Topic.Select(topic);
                publishAction(tp);
                tp.Publish();

                _resets[correlationID].Wait(timeout);

                response = _responses[correlationID] as TResponse;
                exception = _exceptions[correlationID];

                if (exception != null) throw exception;

                return response;
            });
        }

        private static TResponse SendRequest<TResponse>(string topic, ITransport transport, int timeout, Action<Topic> publishAction, Guid correlationID) where TResponse : class {

            SetupRequestResponse<TResponse>(topic, transport, correlationID);

            var response = default(TResponse);
            var exception = _exceptions[correlationID] = default(Exception);

            _responses[correlationID] = response;
            _resets[correlationID] = new ManualResetEventSlim(false);

            if (timeout <= 0) timeout = -1;

            var tp = Topic.Select(topic);
            publishAction(tp);
            tp.Publish();

            _resets[correlationID].Wait(timeout);

            response = _responses[correlationID] as TResponse;
            exception = _exceptions[correlationID];

            if(exception != null) throw exception;
            
            return response;
        }

        public void Handle<TRequest, TResponse>(int interval, ITransport transport, Func<TRequest, TResponse> func)
            where TRequest : class
            where TResponse : class 
        {
            if (_subscriber == null)
                throw new InvalidOperationException("ProxyService.Start must be called to enable IESBMessage<TRequest,TResponse>");


            var key = String.Format("{0}-{1}", typeof(TRequest).FullName, typeof(TResponse).FullName);
            var correlationID = _guidKeys.GetOrAdd(key, _ => Guid.NewGuid());

            var transportName = correlationID.ToString();

            Topic.RegisterIfNotExist(typeof(TRequest).Name);
            Topic.RegisterIfNotExist(typeof(TResponse).Name);

            if (interval <= 0) interval = 100;

            if (!_subscriber.IsTransportRegistered(transportName)) {

                if (!_subscriber.IsSubscribeTo<TRequest>())
                    _subscriber
                        .SubscribeTo(Topic.Select<TRequest>()).Save();

                _subscriber
                    .AddTransportFor<TRequest>(transportName, transport)
                    .Save();
            }

            if (!_handlers.ContainsKey(correlationID)) {
                _subscriber = Subscriber.Select(_subscriber.SubscriberName);
                _handlers[correlationID] =  
                    _subscriber.OnMessageReceived<TRequest>(transportName, msg => {
                        var correlation = msg as ICorrelation;
                        var id = correlation != null ? correlation.ESBCorrelationID : Guid.Empty;
                        var response = func(msg);
                        if (id != Guid.Empty && response is ICorrelation)
                            (response as ICorrelation).ESBCorrelationID = id;
                        Topic.PublishMessage(response);
                    }, error => {
                        error.Continue = true;
                    }, interval: TimeSpan.FromMilliseconds(interval));
            }
        }
    }
}
